package com.alibaba.cobar.route.function;

import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;

import com.alibaba.cobar.config.model.rule.RuleAlgorithm;
import com.alibaba.cobar.parser.ast.expression.Expression;
import com.alibaba.cobar.parser.ast.expression.primary.function.FunctionExpression;

/**
 * partition by account bill number
 * 
 * @author tanw
 *
 */
public class PartitionByBillno extends PartitionFunction implements RuleAlgorithm {
	
	private static long BILLYEPOCH = 1438358400000L;  //时间起点，我们定义为2015-08-01
	private static long DAYMILLS = 24*60*60*1000;	//一天的毫秒数
	
	//历史台帐数据库数量
	private int historyCount;
	//当月台帐数据库数量
	private int currentCount;
	
	public PartitionByBillno(String functionName) {
        this(functionName, null);
    }

    public PartitionByBillno(String functionName, List<Expression> arguments) {
        super(functionName, arguments);
    }

    @Override
    public Object evaluationInternal(Map<? extends Object, ? extends Object> parameters) {
        return calculate(parameters)[0];
    }

    @Override
    public Integer[] calculate(Map<? extends Object, ? extends Object> parameters) {
        Integer[] rst = new Integer[1];
        Object arg = arguments.get(0).evaluation(parameters);
        if (arg == null) {
            throw new IllegalArgumentException("partition key is null ");
        } else if (arg == UNEVALUATABLE) {
            throw new IllegalArgumentException("argument is UNEVALUATABLE");
        }
        Number key;
        if (arg instanceof Number) {
            key = (Number) arg;
        } else if (arg instanceof String) {
            key = Long.parseLong((String) arg);
        } else {
            throw new IllegalArgumentException("unsupported data type for partition key: " + arg.getClass());
        }
        
        String keyStr = Long.toString(key.longValue());
        String keyStr2 = PartitionByBillno.leftPad(keyStr, 12, '0');
        String dayStr = keyStr2.substring(0, 4);
        String idStr = keyStr2.substring(5);

        long billRegDay = BILLYEPOCH+Short.valueOf(dayStr)*DAYMILLS;
        long preMonthFirstDay = getSthMonthFirstDay(-1);
        
        //如果帐单日期大于当前时间超过7天，则认为是非法超前数据
        if(billRegDay > System.currentTimeMillis() && ((billRegDay-System.currentTimeMillis()/DAYMILLS) > 7)){
        	throw new RuntimeException("invalid bill number!! bill regdate greater than today 7 days");
        }
         
        boolean isCurrMonth = true;
        
        //帐单日期大于当前日期的上个月1号定义为历史数据，其他的为当前（最近）数据
        if(billRegDay < preMonthFirstDay){
        	isCurrMonth = false;
        }

        int realKey = 0;
        
        if(isCurrMonth){
        	realKey = partitionIndex(Long.valueOf(idStr));
        }else{//历史数据，数据源索引偏移
        	realKey = currentCount+(partitionIndex(Long.valueOf(idStr)) % historyCount);
        }
        
        rst[0] = realKey;

        return rst;
    }

    @Override
    public FunctionExpression constructFunction(List<Expression> arguments) {
        if (arguments == null || arguments.size() != 1)
            throw new IllegalArgumentException("function " + getFunctionName() + " must have 1 argument but is "
                    + arguments);
        Object[] args = new Object[arguments.size()];
        int i = -1;
        for (Expression arg : arguments) {
            args[++i] = arg;
        }
        return (FunctionExpression) constructMe(args);
    }

    @Override
    public RuleAlgorithm constructMe(Object... objects) {
        List<Expression> args = new ArrayList<Expression>(objects.length);
        for (Object obj : objects) {
            args.add((Expression) obj);
        }
        PartitionByBillno partitionFunc = new PartitionByBillno(functionName, args);
        partitionFunc.count = count;
        partitionFunc.length = length;
        partitionFunc.historyCount = historyCount;
        partitionFunc.currentCount = currentCount;
        return partitionFunc;
    }

    @Override
    public void initialize() {
        init();
    }
    
    public int getHistoryCount() {
		return historyCount;
	}

	public void setHistoryCount(int historyCount) {
		this.historyCount = historyCount;
	}
	

	public int getCurrentCount() {
		return currentCount;
	}

	public void setCurrentCount(int currentCount) {
		this.currentCount = currentCount;
	}

	protected static String leftPad(String str,int len,char ch){
    	StringBuffer ret = new StringBuffer(str);
    	
    	if(str.length()>=len || str.length() <= 0){
    		return ret.toString();
    	}
    	
    	if(str.length()<len){
    		for(int i=0;i<(len-str.length());i++){
    			ret.insert(0, ch);
    		}
    	}
    	
    	return ret.toString();
    }
	
	protected static long diffDays(Date d1,Date d2) {
		long t = d1.getTime()-d2.getTime();
		return t/(24*60*60*1000);
	}
	
	protected static long getSthMonthFirstDay(int month){
		
        Calendar cal = Calendar.getInstance();
        cal.setTime(new Date());
        cal.add(Calendar.MONTH, month);
        cal.set(Calendar.DAY_OF_MONTH,cal.getActualMinimum(Calendar.DAY_OF_MONTH));
        //SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        //System.out.println("当月时间上月1号-->"+sdf.format(cal.getTime())); 
        return cal.getTimeInMillis();
	}
    
}
